package cz.data.common.elastic.manage;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import cz.data.common.core.DataConstant;
import cz.data.common.elastic.CacheClientBean;
import cz.data.common.elastic.ElasticSearchUtils;
import cz.data.common.exception.DataException;
import cz.data.common.support.db.DbProperty;
import cz.data.common.support.db.core.DbColumn;
import cz.data.common.support.db.core.DbTable;
import cz.data.common.support.db.manage.Condition;
import cz.data.common.support.db.manage.DbManage;
import cz.data.common.utils.ThrowableUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.IndicesClient;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateDataStreamRequest;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.xcontent.XContentType;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.stream.Collectors;

@Slf4j
public class EsManage implements DbManage {
    private static final String INDEX_MODEL_KEY = "index_model";
    private static final String DEFAULT_POLICY = "data_platform_index_policy";
    private final RestHighLevelClient rest;
    private final ElasticsearchActuator actuator;

    public EsManage(DbProperty property) {
        this.rest = CacheClientBean.createClient(property);
        this.actuator = new ElasticsearchActuator(this.rest);
    }


    @Override
    public List<DbTable> getTables(String dbName, String schemaName) {
        return actuator.indexNames()
                .stream()
                .map(actuator::tableMapping)
                .collect(Collectors.toList());
    }

    @Override
    public List<DbColumn> getTableColumns(String dbName, String schemaName, String tableName) {
        return actuator.columnMapping(tableName);
    }

    @Override
    public boolean createTable(DbTable table, Properties properties) {
        String indexModel = properties.getProperty(INDEX_MODEL_KEY, DataConstant.ElasticsearchIndexModel.Normal.getValue());
        if (DataConstant.ElasticsearchIndexModel.Normal.equalsValue(indexModel)) {
            return actuator.createIndex(table, properties);
        } else if (DataConstant.ElasticsearchIndexModel.Roll.equalsValue(indexModel)) {
            boolean success = actuator.createDefaultIngestPipeline();
            if (!success)
                return false;
            success = actuator.createLifecyclePolicy(DEFAULT_POLICY);
            if (!success)
                return false;
            success = actuator.createIndexTemplate(table, properties);
            if (!success)
                return false;
            IndicesClient client = this.rest.indices();
            CreateIndexRequest request = new CreateIndexRequest(table.getTableName() + "-000001");
            return Optional.ofNullable(
                            ThrowableUtil.callable(
                                    () -> client.create(request, RequestOptions.DEFAULT), false)
                    ).map(CreateIndexResponse::isAcknowledged)
                    .orElse(false);
        } else if (DataConstant.ElasticsearchIndexModel.Stream.equalsValue(indexModel)) {
            boolean success = actuator.createDefaultIngestPipeline();
            if (!success)
                return false;
            success = actuator.createLifecyclePolicy(DEFAULT_POLICY);
            if (!success)
                return false;
            String dataStreamName = table.getTableName() + "-stream";
            properties.put("data_stream", dataStreamName);
            success = actuator.createIndexTemplate(table, properties);
            if (!success)
                return false;
            IndicesClient client = this.rest.indices();
            return Optional.ofNullable(
                            ThrowableUtil.callable(
                                    () -> client.createDataStream(new CreateDataStreamRequest(dataStreamName),
                                            RequestOptions.DEFAULT), false)
                    ).map(AcknowledgedResponse::isAcknowledged)
                    .orElse(false);
        }
        return false;
    }

    @Override
    public boolean deleteTable(DbTable table) {
        IndicesClient client = this.rest.indices();
        DeleteIndexRequest request = new DeleteIndexRequest(table.getTableName());
        AcknowledgedResponse response = ThrowableUtil.callable(
                () -> client.delete(request, RequestOptions.DEFAULT), false);
        return Optional.ofNullable(response)
                .map(AcknowledgedResponse::isAcknowledged)
                .orElse(false);
    }

    @Override
    public boolean insertData(DbTable table, JSONObject record) {
        JSONObject json = DbManage.filterJson(table.getColumns(), record);
        IndexRequest request = new IndexRequest(table.getTableName());
        String primaryKeys = primaryKeys(table.getColumns(), record);
        request.id(primaryKeys);
        request.source(json.toJSONString(), XContentType.JSON);
        IndexResponse response = ThrowableUtil.callable(() -> rest.index(request, RequestOptions.DEFAULT));
        return Optional.ofNullable(response)
                .map(IndexResponse::getShardInfo)
                .map(ReplicationResponse.ShardInfo::getFailed)
                .map(failed -> failed.equals(0))
                .orElse(false);
    }

    @Override
    public boolean updateData(DbTable table, JSONObject record) {
        return insertData(table, record);
    }

    @Override
    public boolean removeData(DbTable table, JSONObject record) {
        List<DbColumn> columns = table.getColumns();
        List<TermQueryBuilder> termQueryBuilders = new ArrayList<>();
        for (DbColumn column : columns) {
            String colName = column.getColName();
            Object value = record.get(colName);
            if (Objects.nonNull(value)) {
                termQueryBuilders.add(QueryBuilders.termQuery(colName, value));
            }
        }
        if (CollectionUtils.isEmpty(termQueryBuilders))
            throw new DataException("缺少必要的主键参数");
        DeleteByQueryRequest request = new DeleteByQueryRequest(table.getTableName());
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        termQueryBuilders.forEach(boolQuery::filter);
        request.setQuery(boolQuery);
        BulkByScrollResponse response = ThrowableUtil.callable(() -> rest.deleteByQuery(request, RequestOptions.DEFAULT));
        return Optional.ofNullable(response)
                .map(BulkByScrollResponse::getBulkFailures)
                .map(CollectionUtils::isEmpty)
                .orElse(false);
    }

    @Override
    public IPage<JSONObject> getPageData(DbTable table, List<String> fields, List<Condition> conditions, Page<JSONObject> page) {
        List<DbColumn> columns = table.getColumns();
        List<QueryBuilder> queryBuilders = ElasticSearchUtils.parseQueryBuilds(columns, conditions);
        QueryBuilder queryBuilder;
        if (CollectionUtils.isNotEmpty(queryBuilders)) {
            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
            queryBuilders.forEach(boolQuery::filter);
            queryBuilder = boolQuery;
        } else {
            queryBuilder = QueryBuilders.matchAllQuery();
        }
        SearchRequest request = new SearchRequest(table.getTableName());
        SearchSourceBuilder query = new SearchSourceBuilder().query(queryBuilder);
        query.from((int) page.offset());
        query.size((int) page.getSize());
        request.source(query);
        SearchResponse response = ThrowableUtil.callable(() -> rest.search(request, RequestOptions.DEFAULT));
        Optional.ofNullable(response)
                .map(SearchResponse::getHits)
                .ifPresent(hits -> {
                    List<JSONObject> collect = Arrays.stream(hits.getHits())
                            .map(SearchHit::getSourceAsMap)
                            .map(JSONObject::new)
                            .collect(Collectors.toList());
                    page.setRecords(collect);
                    page.setTotal(hits.getTotalHits().value);
                });
        return page;
    }

    @Override
    public JSONObject getDataById(DbTable table, JSONObject params) {
        List<DbColumn> columns = table.getColumns();
        List<QueryBuilder> queryBuilders = new ArrayList<>();
        for (DbColumn column : columns) {
            if (Boolean.TRUE.equals(column.getColKey())) {
                String colName = column.getColName();
                Object value = params.get(colName);
                if (Objects.nonNull(value)) {
                    queryBuilders.add(QueryBuilders.termQuery(colName, value));
                }
            }
        }
        if (CollectionUtils.isEmpty(queryBuilders))
            throw new DataException("缺少必要的主键参数");
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        queryBuilders.forEach(boolQuery::filter);
        SearchRequest request = new SearchRequest(table.getTableName());
        SearchSourceBuilder query = new SearchSourceBuilder().query(boolQuery);
        query.size(1);
        request.source(query);
        SearchResponse response = ThrowableUtil.callable(() -> rest.search(request, RequestOptions.DEFAULT));
        return Optional.ofNullable(response)
                .map(SearchResponse::getHits)
                .flatMap(hits -> Arrays.stream(hits.getHits())
                        .map(SearchHit::getSourceAsMap)
                        .map(JSONObject::new)
                        .findFirst()
                ).orElse(null);
    }

    @Override
    public void close() throws Exception {
//        if (Objects.nonNull(this.rest)) {
//            this.rest.close();
//        }
    }

    public String primaryKeys(List<DbColumn> columns, JSONObject record) {
        StringBuilder sb = new StringBuilder();
        for (DbColumn column : columns) {
            if (Boolean.TRUE.equals(column.getColKey())) {
                String colName = column.getColName();
                Object value = record.get(colName);
                if (Objects.nonNull(value)) {
                    if (value instanceof Date) {
                        sb.append(((Date) value).getTime());
                    } else if (value instanceof LocalDateTime) {
                        sb.append(((LocalDateTime) value).getSecond());
                    } else {
                        sb.append(value);
                    }
                }
            }
        }
        return sb.length() > 0 ? sb.toString() : null;
    }
}
